package com.jason.handler;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.kafka.listener.KafkaListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.Collection;

@Component("consumerErrorHandler")
@Slf4j
public class ConsumerErrorHandler implements KafkaListenerErrorHandler {

    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception) {
        log.info("~~~~~~~~消费出错,用户组id{},错误信息:{}",exception.getGroupId(),exception.getMessage());
        return null;
    }

    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
        log.info("~~~~default~~~~消费出错,用户组id{},错误信息:{},主题:{}",exception.getGroupId(),exception.getMessage());
        return null;
    }
}
